package com.fenghuaxz.rpcframework;

import com.fenghuaxz.rpcframework.annotation.Redefine;
import com.fenghuaxz.rpcframework.model.MsgCallback;
import com.fenghuaxz.rpcframework.model.MsgRequest;
import com.fenghuaxz.rpcframework.model.SyncCallback;
import com.fenghuaxz.rpcframework.model.UnsyncCallback;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.channel.Channel;
import io.netty.channel.DefaultChannelId;
import io.netty.util.ReferenceCountUtil;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
/**
 * 消息包裹
 * 创建 : 七羽软件工作室
 * 时间 : 2017/9/28 : 2:25
 */
public final class MessagePacket {

    private Channel[] channel = null;
    private BroadcastContext context = null;
    private OnCallbackListener listener = null;

    public MessagePacket(BroadcastContext context, Channel... channel) {
        this.context = context;
        this.channel = channel;
    }

    public final MessagePacket setCallback(OnCallbackListener listener){
        this.listener = listener;
        return this;
    }

    public final synchronized <T> T getBean(Class<T> clazz){
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                MsgRequest request = wrapRequest(method, args);
                MsgCallback callback = listener != null ? new UnsyncCallback(method,listener) : new SyncCallback();
                return callback instanceof SyncCallback ? writeAndFlushSyncRequest(request,callback) : writeAndFlushUnsyncRequest(request,callback);
            }
        });
    }

    /**
     * 发起异步请求
     * @param request
     * @param callback
     * @return
     * @throws Exception
     */
    private final Object writeAndFlushUnsyncRequest(MsgRequest request,MsgCallback callback) throws Exception{
        for (Channel c : channel){
            request.setMsgId(randomRequestId());
            MessagePacket.this.context.callbacks.put(request.getMsgId(),callback);
            c.writeAndFlush(safeDuplicate(request));
        }
        return callback.await();
    }

    /**
     * 发起同步请求
     * @param request
     * @param callback
     * @return
     * @throws Exception
     */
    private final Object writeAndFlushSyncRequest(MsgRequest request,MsgCallback callback) throws Exception{
        if (channel.length != 1)
            throw new UnsupportedOperationException(String.format("the sync request pipe family size must be 1 ,current %d",channel.length));
        else{
            MessagePacket.this.context.callbacks.put(request.getMsgId(),callback);
            channel[0].writeAndFlush(safeDuplicate(request));
            return callback.await();
        }
    }

    private final String randomRequestId(){
        return DefaultChannelId.newInstance().asShortText();
    }

    private final MsgRequest wrapRequest(Method method,Object[] args){
        Redefine redefine = method.getDeclaringClass().getAnnotation(Redefine.class);
        String className = redefine != null ? redefine.name() : method.getDeclaringClass().getName();
        Redefine.Method remethod = method.getAnnotation(Redefine.Method.class);
        String methodName = remethod != null ? remethod.name() : method.getName();
        MsgRequest request = new MsgRequest();
        request.setMsgId(randomRequestId());
        request.setClassName(className);
        request.setMethodName(methodName);
        request.setParamTypes(method.getParameterTypes());
        request.setParams(args);
        return request;
    }

    private static Object safeDuplicate(Object message) {
        if (message instanceof ByteBuf) {
            return ((ByteBuf)message).retainedDuplicate();
        } else {
            return message instanceof ByteBufHolder ? ((ByteBufHolder)message).retainedDuplicate() : ReferenceCountUtil.retain(message);
        }
    }
}